package com.deep.stream.service;

import com.deep.stream.models.OrganizationChangeModel;
import com.deep.stream.utils.UserContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @author dell
 * @create 2022/7/29 9:19
 */

// 为了简洁，省略了import语句

@EnableBinding(value = {SimpleSource.class})
@Component
public class SimpleSourceBean {
    private SimpleSource simpleSource;
    private Source source;

    private static final Logger logger = LoggerFactory.getLogger(SimpleSourceBean.class);

    @Autowired
    public SimpleSourceBean(SimpleSource simpleSource, Source source) {
        // Spring Cloud Stream将注入一个Source接口，以供服务使用
        this.simpleSource = simpleSource;
        this.source = source;
    }

    public void publish(String msg) {
        logger.debug("publish Kafka msg {}for msg: {}", msg, msg);
        source.output().send(MessageBuilder.withPayload(msg).build());
    }

    public void publishOrgChange(String action, String orgId) {
        logger.debug("Sending Kafka message {}for Organization Id: {}",
                action, orgId);
        OrganizationChangeModel change = new OrganizationChangeModel(
                OrganizationChangeModel.class.getTypeName(),
                action,
                orgId,
                UserContext.getCorrelationId()); // 要发布的消息是一个Java POJO

        simpleSource.output().send(MessageBuilder.withPayload(change).build()); // 当准备发送消息时，使用Source类中定义的通道的send()方法
    }
}
